package com.yanqu.road.task;

import com.yanqu.road.message.YanQuMessage;
import com.yanqu.road.message.YanQuMessageBuffer;
import com.yanqu.road.utils.YanQuMessageUtils;
import com.yanqu.road.connect.YanQuConnection;
import io.netty.channel.ChannelPromise;

public class FlushConnectionTask implements Runnable{
    private YanQuConnection connection;

    public FlushConnectionTask(YanQuConnection connection) {
        this.connection = connection;
    }

    @Override
    public void run() {
        connection.getFlushSessionTaskSwitch().set(true);
        if (connection.getMessageQueue().size() == 1) {
            YanQuMessage yanQuMessage = connection.getMessageQueue().poll();
            if (yanQuMessage != null) {
                ChannelPromise promise = newPromise();
                connection.getChannel().writeAndFlush(yanQuMessage, promise);
            }
        } else if (connection.getMessageQueue().size() > 1) {
            YanQuMessageBuffer buffer = new YanQuMessageBuffer();
            connection.getMessageQueue().drainTo(buffer);
            if (!buffer.isEmpty()) {
                YanQuMessage msg = YanQuMessageUtils.buildMessage((short) 0, buffer);
                ChannelPromise promise = newPromise();
                connection.getChannel().writeAndFlush(msg, promise);
            }
        }
    }

    private ChannelPromise newPromise() {
        ChannelPromise promise = connection.getChannel().newPromise();
        promise.addListener(connection.getListener());
        connection.getCounter().incrementAndGet();
        return promise;
    }
}
